package com.producer.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import java.lang.reflect.Type;
import java.util.List;

/**
 * @author Liunh
 */
@Slf4j
public class SendMsgUtil {

    @Value("${rocketmq.producer.send-message-timeout:15000}")
    private Integer messageTimeOut;
    private final RocketMQTemplate rocketMqTemplate;

    public SendMsgUtil(RocketMQTemplate rocketMqTemplate) {
        this.rocketMqTemplate = rocketMqTemplate;
    }


    /**
     * 发送事务消息
     *
     * @param msgBody
     * @param topic
     * @param tag
     */
    public void sendTransactionMsg(final String msgBody, final String topic, final String tag) {
        log.info(">>>>>> mq provider send transaction msg");
        final Message<String> message = MessageBuilder.withPayload(msgBody).build();
        final TransactionSendResult res = rocketMqTemplate.sendMessageInTransaction(topic, message, tag);
        if (res.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE) && res.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info(">>>>>> 事务消息发送成功");
        }
    }

    /**
     * 发送消息
     * 返回值是string
     */
    public SendResult sendAndReceive(final Message msgBody, final String topic) {
        log.info(">>>>>> mq provider send batch msg");
        return rocketMqTemplate.sendAndReceive(topic, msgBody, String.class);
    }

    /**
     * 批量消息
     * 返回值是[type]
     */
    public SendResult sendAndReceive(final Message msgBody, final String topic, final Type type) {
        log.info(">>>>>> mq provider send batch msg");
        return rocketMqTemplate.sendAndReceive(topic, msgBody, type);
    }

    /**
     * 批量发送消息
     * 批量消息体不能大于4MB
     */
    public SendResult sendMsgBatch(final List msgBody, final String topic) {
        log.info(">>>>>> mq provider send batch msg");
        return rocketMqTemplate.syncSend(topic, msgBody);
    }

    /**
     * 普通发送（这里的参数对象User可以随意定义，可以发送个对象，也可以是字符串等）
     */
    public boolean sendMsg(final String msgBody, final String topic) {
        boolean result = true;
        try {
            log.info(">>>>>> mq provider send msg");
            rocketMqTemplate.convertAndSend(topic, msgBody);
            //mqTemplate.send(topic, MessageBuilder.withPayload(user).build());
        } catch (Exception e) {
            result = false;
            log.error(">>>>>> RocketMqUtil sendMsg error!", e);
        }
        return result;
    }

    /**
     * 发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
     * （msgBody也可以是对象，sendResult为返回的发送结果）
     */
    public SendResult sendSyncMsg(final String msgBody, final String topic) {
        log.info(">>>>>> mq provider send msg");
        return rocketMqTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build());
    }

    /**
     * 普通发送（这里的参数对象User可以随意定义，可以发送个对象，也可以是字符串等）
     * 设置 Tag：用于区分过滤同一主题下的不同业务类型的消息，非常实用
     * 在项目里往mq写入消息时，最好每条消息都带上tag，用于消费时根据业务过滤
     */
    public boolean sendMsg(final String msgBody, final String topic, final String tag) {
        boolean result = true;
        try {
            log.info(">>>>>> mq provider send msg");
            rocketMqTemplate.convertAndSend(topic + ":" + tag, msgBody);
            //mqTemplate.send(topic, MessageBuilder.withPayload(user).build());
        } catch (Exception e) {
            result = false;
            log.error(">>>>>> RocketMqUtil sendMsg error!", e);
        }
        return result;
    }

    /**
     * 普通发送（这里的参数对象User可以随意定义，可以发送个对象，也可以是字符串等）
     * 设置 Tag：用于区分过滤同一主题下的不同业务类型的消息，非常实用
     * 在项目里往mq写入消息时，最好每条消息都带上tag，用于消费时根据业务过滤
     */
    public boolean sendMsg(final Message msgBody, final String topic, final String tag) {
        boolean result = true;
        try {
            log.info(">>>>>> mq provider send msg");
            rocketMqTemplate.convertAndSend(topic + ":" + tag, msgBody);
        } catch (Exception e) {
            result = false;
            log.error(">>>>>> RocketMqUtil sendMsg error!", e);
        }
        return result;
    }

    /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     */
    public SendResult sendSyncMsg(final String msgBody, final String topic, final String tag) {
        log.info(">>>>>> mq provider send sync msg");
        return rocketMqTemplate.syncSend(topic + ":" + tag, MessageBuilder.withPayload(msgBody).build());
    }

    /**
     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
     * （适合对响应时间敏感的业务场景）
     */
    public boolean sendAsyncMsg(final String msgBody, final String topic) {
        log.info(">>>>>> mq provider send async msg");
        final boolean[] result = {true};
        rocketMqTemplate.asyncSend(topic, MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info(">>>>>> mq provider send async msg success !");
                // 处理消息发送成功逻辑
            }

            @Override
            public void onException(Throwable throwable) {
                log.error(">>>>>> mq provider send async msg false !");
                result[0] = false;
                // 处理消息发送异常逻辑
            }
        });
        return result[0];
    }

    /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public boolean sendDelayMsg(final String msgBody, final String topic, final int delayLevel) {
        log.info(">>>>>> mq provider send delay msg");
        boolean result = true;
        try {
            rocketMqTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build(), messageTimeOut, delayLevel);
        } catch (Exception e) {
            result = false;
            log.error(">>>>>> RocketMqUtil sendDelayMsg error!", e);
        }
        return result;
    }

    /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public boolean sendDelayMsg(final String msgBody, final String topic, final Integer timeOut, final int delayLevel) {
        log.info(">>>>>> mq provider send delay msg");
        boolean result = true;
        try {
            rocketMqTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build(), timeOut, delayLevel);
        } catch (Exception e) {
            result = false;
            log.error(">>>>>> RocketMqUtil sendDelayMsg error!", e);
        }
        return result;
    }

    /**
     * 发送单向消息（顺序消息,只负责发送消息，不等待应答，不关心发送结果，如日志）
     */
    public boolean sendOneWayMsg(final String msgBody, final String topic) {
        log.info(">>>>>> mq provider send one way msg");
        boolean result = true;
        try {
            rocketMqTemplate.sendOneWay(topic, MessageBuilder.withPayload(msgBody).build());
        } catch (Exception e) {
            result = false;
            log.error(">>>>>> RocketMqUtil sendOneWayMsg error!", e);
        }
        return result;
    }


}
